/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.gateway.register.client.server.nacos;

import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.config.ConfigFactory;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.Listener;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.gateway.common.constant.Constants;
import com.gateway.common.constant.NacosPathConstants;
import com.gateway.common.enums.RpcTypeEnum;
import com.gateway.common.exception.GatewayException;
import com.gateway.common.utils.JsonUtils;
import com.gateway.register.client.server.api.GatewayClientServerRegisterPublisher;
import com.gateway.register.client.server.api.GatewayClientServerRegisterRepository;
import com.gateway.register.common.config.GatewayRegisterCenterConfig;
import com.gateway.register.common.dto.MetaDataRegisterDTO;
import com.gateway.register.common.dto.RouteRegisterDTO;
import com.gateway.register.common.path.RegisterPathConstants;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executor;

/**
 * Nacos register server.
 */
public class NacosClientServerRegisterRepository implements GatewayClientServerRegisterRepository {

    private static final Logger log = LoggerFactory.getLogger(NacosClientServerRegisterRepository.class);

    private final String defaultGroup = NacosPathConstants.GROUP;

    private ConfigService configService;

    private NamingService namingService;

    private GatewayClientServerRegisterPublisher publisher;

    private final ConcurrentSkipListSet<String> metadataConfigCache = new ConcurrentSkipListSet<>();
    private final ConcurrentMap<String, ConcurrentSkipListSet<String>> metaServiceCache = new ConcurrentHashMap<>();

    private final ConcurrentMap<String, ConcurrentSkipListSet<String>> routeServiceCache = new ConcurrentHashMap<>();

    @Override
    public void close() {
        publisher.close();
    }

    @Override
    public String registerType() {
        return "nacos";
    }

    @Override
    public void init(final GatewayClientServerRegisterPublisher publisher,
                     final GatewayRegisterCenterConfig config) {
        this.publisher = publisher;
        String serverAddr = config.getServerLists();
        Properties properties = config.getProps();
        Properties nacosProperties = new Properties();
        nacosProperties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
        nacosProperties.put(PropertyKeyConst.NAMESPACE, properties.getProperty("nacosNameSpace"));
        // the nacos authentication username
        nacosProperties.put(PropertyKeyConst.USERNAME, properties.getProperty(PropertyKeyConst.USERNAME, ""));
        // the nacos authentication password
        nacosProperties.put(PropertyKeyConst.PASSWORD, properties.getProperty(PropertyKeyConst.PASSWORD, ""));
        // access key for namespace
        nacosProperties.put(PropertyKeyConst.ACCESS_KEY, properties.getProperty(PropertyKeyConst.ACCESS_KEY, ""));
        // secret key for namespace
        nacosProperties.put(PropertyKeyConst.SECRET_KEY, properties.getProperty(PropertyKeyConst.SECRET_KEY, ""));

        try {
            this.configService = ConfigFactory.createConfigService(nacosProperties);
            this.namingService = NamingFactory.createNamingService(nacosProperties);
        } catch (NacosException e) {
            throw new GatewayException(e);
        }

        subscribe();
    }
    /**
     * 目前只接入SpringCloud
     */
    private void subscribe() {
        subscribeRpcTypeService(RpcTypeEnum.SPRING_CLOUD);
    }

    private void subscribeRpcTypeService(final RpcTypeEnum rpcType) {
        final String serviceName = RegisterPathConstants.buildServiceInstancePath(rpcType.getName());
        try {
            Map<String, RouteRegisterDTO> services = Maps.newHashMap();
            List<Instance> healthyInstances = namingService.selectInstances(serviceName, this.defaultGroup,true);
            healthyInstances.forEach(healthyInstance -> {
                String contextPath = healthyInstance.getMetadata().get(Constants.CONTEXT_PATH);
                String serviceConfigName = RegisterPathConstants.buildServiceConfigPath(rpcType.getName(), contextPath);
                subscribeMetaData(serviceConfigName);
                metadataConfigCache.add(serviceConfigName);
                String routeMetadata = healthyInstance.getMetadata().get(Constants.ROUTE_META_DATA);
                RouteRegisterDTO routeRegisterDTO = JsonUtils.stringToBean(routeMetadata, RouteRegisterDTO.class);
                assert routeRegisterDTO != null;
                routeRegisterDTO.setContextPath(contextPath);
                if (!services.containsKey(contextPath)) {
                    services.put(contextPath, routeRegisterDTO);
                }
                routeServiceCache.computeIfAbsent(serviceName, k -> new ConcurrentSkipListSet<>()).add(contextPath);
            });
            if (RpcTypeEnum.acquireSupportRoutes().contains(rpcType)) {
                services.values().forEach(this::publishRegisterRoute);
            }
            log.info("subscribe route : {}", serviceName);
            // 刷新元数据/路由
            namingService.subscribe(serviceName, this.defaultGroup, event -> {
                if (event instanceof NamingEvent) {
                    List<Instance> instances = ((NamingEvent) event).getInstances();
                    if (CollectionUtils.isEmpty(instances)) {
                        return;
                    }
                    for (Instance instance : instances) {
                        String contextPath = instance.getMetadata().get(Constants.CONTEXT_PATH);
                        routeServiceCache.computeIfAbsent(serviceName, k -> new ConcurrentSkipListSet<>()).add(contextPath);
                    }
                    refreshRouteService(rpcType, serviceName);
                }
            });
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    private void refreshRouteService(final RpcTypeEnum rpcType, final String serviceName) {
        Optional.ofNullable(routeServiceCache.get(serviceName)).ifPresent(services -> services.forEach(contextPath -> refreshRoute(contextPath, rpcType, serviceName)));
    }

    private void refreshRoute(final String contextPath, final RpcTypeEnum rpcType, final String serviceName) {
        try {
            List<Instance> healthyInstances = namingService.selectInstances(serviceName, this.defaultGroup,true);
            if (CollectionUtils.isEmpty(healthyInstances)) {
                return;
            }
            RouteRegisterDTO routeRegisterDTO = null;
            for (Instance healthyInstance : healthyInstances) {
                if (org.apache.commons.lang3.StringUtils.equals(contextPath, healthyInstance.getMetadata().get(Constants.CONTEXT_PATH))) {
                    String routeMetadata = healthyInstance.getMetadata().get(Constants.ROUTE_META_DATA);
                    routeRegisterDTO = JsonUtils.stringToBean(routeMetadata, RouteRegisterDTO.class);
                    String serviceConfigName = RegisterPathConstants.buildServiceConfigPath(rpcType.getName(), contextPath);
                    if (!metadataConfigCache.contains(serviceConfigName)) {
                        subscribeMetaData(serviceConfigName);
                        metadataConfigCache.add(serviceConfigName);
                    }
                }
            }
            if (!RpcTypeEnum.acquireSupportRoutes().contains(rpcType)) {
                return;
            }
            if (Objects.isNull(routeRegisterDTO)) {
//                routeRegisterDTO = RouteRegisterDTO.Builder.builder()
//                        .appName(contextPath)
//                        .contextPath(contextPath)
//                        .serviceId(contextPath)
//                        .rpcType(rpcType.getName())
//                        .env(this.environment.getProperty(GatewayConstants.ENV))
//                        .build();
                return;
            }
            publishRegisterRoute(routeRegisterDTO);
            log.info("subscribe route : {}", serviceName);
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    private void subscribeRpcTypeRouteService(final RpcTypeEnum rpcType) {
        final String serviceName = RegisterPathConstants.buildServiceInstancePath(rpcType.getName());
        Map<String, RouteRegisterDTO> services = Maps.newHashMap();
        try {
            List<Instance> healthyInstances = namingService.selectInstances(serviceName, this.defaultGroup,true);
            healthyInstances.forEach(healthyInstance -> {
                String contextPath = healthyInstance.getMetadata().get(Constants.CONTEXT_PATH);
                String routeMetadata = healthyInstance.getMetadata().get(Constants.ROUTE_META_DATA);
                RouteRegisterDTO routeRegisterDTO = JsonUtils.stringToBean(routeMetadata, RouteRegisterDTO.class);
                if (!services.containsKey(contextPath)) {
                    services.put(contextPath, routeRegisterDTO);
                }
            });
            log.info("subscribe route : {}", serviceName);
            if (RpcTypeEnum.acquireSupportRoutes().contains(rpcType)) {
                services.values().forEach(this::publishRegisterRoute);
            }
            // 刷新元数据
            namingService.subscribe(serviceName, this.defaultGroup, event -> {
                if (event instanceof NamingEvent) {
                    List<Instance> instances = ((NamingEvent) event).getInstances();
                    instances.forEach(instance -> {
                        String contextPath = instance.getMetadata().get(Constants.CONTEXT_PATH);
                        routeServiceCache.computeIfAbsent(serviceName, k -> new ConcurrentSkipListSet<>()).add(contextPath);
                    });
                    refreshRouteService(rpcType, serviceName);
                }
            });
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    private void subscribeRpcTypeMetadataService(final RpcTypeEnum rpcType) {
        final String serviceName = RegisterPathConstants.buildServiceInstancePath(rpcType.getName());
        try {
            List<Instance> healthyInstances = namingService.selectInstances(serviceName, this.defaultGroup,true);
            healthyInstances.forEach(healthyInstance -> {
                String contextPath = healthyInstance.getMetadata().get(Constants.CONTEXT_PATH);
                String serviceConfigName = RegisterPathConstants.buildServiceConfigPath(rpcType.getName(), contextPath);
                subscribeMetaData(serviceConfigName);
                metadataConfigCache.add(serviceConfigName);
                metaServiceCache.computeIfAbsent(serviceName, k -> new ConcurrentSkipListSet<>()).add(contextPath);
            });
            // 订阅实例, 刷新元数据
            namingService.subscribe(serviceName, this.defaultGroup, event -> {
                if (event instanceof NamingEvent) {
                    List<Instance> instances = ((NamingEvent) event).getInstances();
                    instances.forEach(instance -> {
                        String contextPath = instance.getMetadata().get(Constants.CONTEXT_PATH);
                        metaServiceCache.computeIfAbsent(serviceName, k -> new ConcurrentSkipListSet<>()).add(contextPath);
                    });
                    refreshMetadataService(rpcType, serviceName);
                }
            });
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }



    private void subscribeMetaData(final String serviceConfigName) {
        String content = readData(serviceConfigName);
        if (StringUtils.isEmpty(content)) {
            return;
        }
        registerMetadata(content);
        log.info("subscribe metadata: {}", serviceConfigName);
        try {
            // 添加监听器，监听配置变动
            configService.addListener(serviceConfigName, NacosPathConstants.GROUP, new Listener() {
                @Override
                public Executor getExecutor() {
                    return null;
                }

                @Override
                public void receiveConfigInfo(final String config) {
                    registerMetadata(config);
                }
            });
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }

    }

    private void registerMetadata(final String metadataConfig) {
        List<String> metadataList = JsonUtils.jsonToList(metadataConfig, String.class);
        metadataList.forEach(this::publishMetadata);
    }

    private void publishMetadata(final String data) {
        log.info("publish metadata: {}", data);
        publisher.publish(Lists.newArrayList(JsonUtils.stringToBean(data, MetaDataRegisterDTO.class)));
    }

    private void refreshMetadataService(final RpcTypeEnum rpcType, final String serviceName) {
        Optional.ofNullable(metaServiceCache.get(serviceName)).ifPresent(services -> services.forEach(contextPath -> registerMetadata(contextPath, serviceName, rpcType)));
    }

    private void registerMetadata(final String contextPath, final String serviceName, final RpcTypeEnum rpcType) {
        try {
            List<Instance> healthyInstances = namingService.selectInstances(serviceName,  this.defaultGroup, true);
            healthyInstances.forEach(healthyInstance -> {
                if (contextPath.equals(healthyInstance.getMetadata().get(Constants.CONTEXT_PATH))) {
                    String serviceConfigName = RegisterPathConstants.buildServiceConfigPath(rpcType.getName(), contextPath);
                    if (!metadataConfigCache.contains(serviceConfigName)) {
                        subscribeMetaData(serviceConfigName);
                        metadataConfigCache.add(serviceConfigName);
                    }
                }
            });
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    private void publishRegisterRoute(final RouteRegisterDTO routeRegisterDTO) {
        log.info("publish route: {}", routeRegisterDTO);
        publisher.publish(routeRegisterDTO);
    }

    private String readData(final String configName) {
        try {
            return configService.getConfig(configName, NacosPathConstants.GROUP, 5000);
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }
}
